package com.darcytech.debezium.core.rpc;

import com.darcytech.debezium.common.utils.ObjectMapperUtils;
import com.darcytech.debezium.core.api.Api;
import com.darcytech.debezium.core.api.RequestHandler;
import com.darcytech.debezium.core.api.bo.CCBindingHolder;
import com.darcytech.debezium.core.pool.ConnectorConnectionBinding;
import com.darcytech.debezium.core.rpc.proto.Rpc;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RpcServerHandler extends SimpleChannelInboundHandler<Rpc.Req> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcServerHandler.class);

    private final ConnectorConnectionBinding connectorConnectionBinding;

    public RpcServerHandler(ConnectorConnectionBinding connectorConnectionBinding) {
        this.connectorConnectionBinding = connectorConnectionBinding;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        LOGGER.warn("{} has been no message for 120 seconds",ctx.channel().remoteAddress());
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                LOGGER.warn("disconnect with {}",ctx.channel().remoteAddress());
                ctx.channel().close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Rpc.Req req) {
        Class<?> requestClazz = Api.Binding.getRequestClazzByType(req.getType());
        Object request = ObjectMapperUtils.readValue(req.getData(), requestClazz);

        setCCBinding(request);
        doHandle(ctx, req, request);
    }

    private void setCCBinding(Object request) {
        if (request instanceof CCBindingHolder) {
            CCBindingHolder cCBindingHolder = (CCBindingHolder) request;
            cCBindingHolder.setConnectorConnectionBinding(connectorConnectionBinding);
        }
    }

    private void doHandle(ChannelHandlerContext ctx, Rpc.Req req, Object request) {
        if (request instanceof RequestHandler) {
            RequestHandler handler = (RequestHandler) request;
            Rpc.Resp resp = Rpc.Resp.newBuilder()
                    .setRequestId(req.getId())
                    .setRequestType(req.getType())
                    .setData(ObjectMapperUtils.writeAsString(handler.handle()))
                    .build();
            ctx.writeAndFlush(resp);
        }
    }

}
